import json
import random
import time
from datetime import datetime, timedelta
from faker import Faker
from kafka import KafkaProducer
from concurrent.futures import ThreadPoolExecutor

# 初始化Faker和Kafka生产者
fake = Faker('zh_CN')
producer = KafkaProducer(
    bootstrap_servers=['hadoop102:9092'],
    value_serializer=lambda x: json.dumps(x, ensure_ascii=False).encode('utf-8')
)

# 配置参数
NUM_USERS = 1000  # 模拟用户数量
NUM_ITEMS = 5000  # 模拟商品数量
DAYS_OF_DATA = 180  # 模拟数据时间范围(天)

# 商品类目和关键词配置(根据任务工单中的母婴类目设计)
MATERNAL_CATEGORIES = {
    # 孕早期
    "孕早期": ["孕妇装", "防辐射服", "孕妇裤", "待产包", "叶酸", "钙铁锌"],
    # 孕中期
    "孕中期": ["孕妇护肤", "孕妇奶粉", "胎教产品", "孕妇内衣"],
    # 孕晚期
    "孕晚期": ["婴儿床", "奶瓶", "吸奶器", "乳头霜", "月子餐"],
    # 0-3个月
    "0-3个月": ["婴儿纸尿裤", "婴儿湿巾", "婴儿洗护", "连身衣", "婴儿帽"],
    # 3-6个月
    "3-6个月": ["婴儿辅食", "米粉", "果泥", "安全座椅", "安抚玩具"],
    # 6-12个月
    "6-12个月": ["学步鞋", "儿童奶粉", "DHA", "益生菌", "早教玩具"],
    # 12-24个月
    "12-24个月": ["幼儿服装", "学步鞋", "幼儿零食", "调味品", "学步车"],
    # 24-36个月
    "24-36个月": ["益智玩具", "幼儿外套", "绘本", "学饮杯"],
    # 3-6岁
    "3-6岁": ["儿童T恤", "儿童裤子", "拼图", "积木", "儿童读物"],
    # 6-12岁
    "6-12岁": ["运动鞋", "文具", "早教机", "儿童零食", "儿童护肤"]
}

# 性别相关关键词(根据任务工单中的性别判断规则设计)
GENDER_KEYWORDS = {
    "男": ["男童", "男宝", "男孩", "男婴", "男宝宝", "男款"],
    "女": ["女童", "女宝", "女孩", "女婴", "女宝宝", "女款"],
    "中性": ["婴儿", "宝宝", "通用", "中性", "男女", "儿童"]
}


# 生成用户基础信息
# 生成用户基础信息（按注册时间升序）
def generate_user_profiles():
    users = []
    base_time = datetime.now() - timedelta(days=2 * 365)  # 2年前开始
    for i in range(NUM_USERS):
        # 每个用户注册时间递增1-3天
        register_time = base_time + timedelta(days=random.randint(1, 3))
        base_time = register_time

        user_id = f"{i:04d}"
        gender = random.choice([0, 1, 2])
        age_range = random.choice(["18-24", "25-30", "31-35", "36-40", "41-45", "46-50"])
        province = fake.province()
        city = fake.city()
        is_parent = random.choices([1, 0], weights=[0.7, 0.3])[0]

        user = {
            "user_id": user_id,
            "register_time": register_time.strftime("%Y-%m-%d %H:%M:%S"),
            "gender": gender,
            "age_range": age_range,
            "province": province,
            "city": city,
            "is_parent": is_parent
        }
        users.append(user)
        producer.send('user_profile_topic', value=user)
    return users


# 生成商品属性表（按创建时间升序）
def generate_item_attributes():
    items = []
    base_time = datetime.now() - timedelta(days=365)  # 1年前开始
    for i in range(NUM_ITEMS):
        # 每个商品创建时间递增1-6小时
        create_time = base_time + timedelta(hours=random.uniform(1, 6))
        base_time = create_time

        item_id = f"{i:05d}"
        item_name = fake.catch_phrase()
        is_maternal = random.choices([0, 1], weights=[0.7, 0.3])[0]

        if is_maternal:
            age_stage = random.choice(list(MATERNAL_CATEGORIES.keys()))
            category_name = random.choice(MATERNAL_CATEGORIES[age_stage])
            category_path = f"母婴用品/{age_stage}/{category_name}"
            gender = random.choices(["男", "女", "中性"], weights=[0.4, 0.4, 0.2])[0]
            keywords = ", ".join(random.sample(GENDER_KEYWORDS[gender], 2))
            description = f"专为{age_stage}{gender if gender != '中性' else ''}宝宝设计的{category_name}"
        else:
            category_path = fake.file_path(depth=3, category=None, extension=None).replace("/", ">")
            keywords = ", ".join(fake.words(3))
            description = fake.sentence()

        item = {
            "item_id": item_id,
            "item_name": item_name,
            "category_id": f"cat_{random.randint(1000, 9999)}",
            "category_path": category_path,
            "keywords": keywords,
            "description": description,
            "price": round(random.uniform(10, 1000), 2),
            "is_maternal": is_maternal,
            "create_time": create_time.strftime("%Y-%m-%d %H:%M:%S")  # 新增创建时间字段
        }
        items.append(item)
        producer.send('item_attributes_topic', value=item)
    return items


# 生成用户行为日志（按行为时间升序）
def generate_user_behavior_logs(users, items):
    behaviors = []
    # 为每个用户创建独立的时间线
    for user in users:
        base_time = datetime.strptime(user["register_time"], "%Y-%m-%d %H:%M:%S")
        num_records = random.randint(10, 50)

        for _ in range(num_records):
            # 行为时间递增5分钟到3天
            behavior_time = base_time + timedelta(minutes=random.randint(5, 4320))
            base_time = behavior_time

            log_id = f"{random.randint(100000, 999999)}"
            item = random.choice(items)
            behavior_type = random.choices(
                ["purchase", "browse", "collect", "add_cart", "search"],
                weights=[0.1, 0.6, 0.1, 0.1, 0.1]
            )[0]

            search_keyword = None
            if behavior_type == "search":
                if item["is_maternal"]:
                    age_stage = item["category_path"].split("/")[1] if "母婴用品" in item["category_path"] else None
                    gender = random.choice(["男", "女", "中性"])
                    search_keyword = f"{age_stage} {gender} {item['item_name'].split()[0]}"
                else:
                    search_keyword = " ".join(fake.words(2))

            behavior = {
                "log_id": log_id,
                "user_id": user["user_id"],
                "item_id": item["item_id"],
                "behavior_type": behavior_type,
                "behavior_time": behavior_time.strftime("%Y-%m-%d %H:%M:%S"),
                "search_keyword": search_keyword,
                "session_id": f"sess_{random.randint(1000, 9999)}",
                "stay_duration": random.randint(1000, 60000),
                "page_url": f"https://example.com/products/{item['item_id']}"
            }
            behaviors.append(behavior)
            producer.send('user_behavior_topic', value=behavior)

    # 按行为时间全局排序
    behaviors.sort(key=lambda x: x["behavior_time"])
    return behaviors


# 生成订单交易数据（按购买时间升序）
def generate_order_transactions(users, items, behaviors):
    orders = []
    purchase_behaviors = [b for b in behaviors if b["behavior_type"] == "purchase"]

    # 按行为时间排序
    purchase_behaviors.sort(key=lambda x: x["behavior_time"])

    for behavior in purchase_behaviors:
        order_id = f"{random.randint(100000, 999999)}"
        purchase_time = datetime.strptime(behavior["behavior_time"], "%Y-%m-%d %H:%M:%S")

        # 确保订单时间不早于商品创建时间
        item_create_time = datetime.strptime(items_dict[behavior["item_id"]]["create_time"], "%Y-%m-%d %H:%M:%S")
        if purchase_time < item_create_time:
            purchase_time = item_create_time + timedelta(minutes=random.randint(10, 1440))

        status = random.choices([1, 2], weights=[0.9, 0.1])[0]

        order = {
            "order_id": order_id,
            "user_id": behavior["user_id"],
            "item_id": behavior["item_id"],
            "purchase_time": purchase_time.strftime("%Y-%m-%d %H:%M:%S"),
            "quantity": random.randint(1, 5),
            "amount": round(float(items_dict[behavior["item_id"]]["price"]) * random.randint(1, 5), 2),
            "payment_method": random.choice(["支付宝", "微信支付", "银行卡", "花呗"]),
            "status": status
        }
        orders.append(order)
        producer.send('order_transaction_topic', value=order)

    return orders


if __name__ == "__main__":
    print("开始生成按时间升序的模拟数据...")

    # 生成用户数据（注册时间升序）
    print("生成用户基础信息（按注册时间升序）...")
    users = generate_user_profiles()
    print(f"已生成 {len(users)} 条用户数据")

    # 生成商品数据（创建时间升序）
    print("生成商品属性数据（按创建时间升序）...")
    items = generate_item_attributes()
    print(f"已生成 {len(items)} 条商品数据")
    items_dict = {item["item_id"]: item for item in items}

    # 生成用户行为数据（行为时间升序）
    print("生成用户行为日志（按行为时间升序）...")
    behaviors = generate_user_behavior_logs(users, items)
    print(f"已生成 {len(behaviors)} 条行为日志")

    # 生成订单数据（购买时间升序）
    print("生成订单交易数据（按购买时间升序）...")
    orders = generate_order_transactions(users, items, behaviors)
    print(f"已生成 {len(orders)} 条订单数据")

    producer.flush()
    print("所有按时间升序的数据已成功写入Kafka")